package org.wolfengi.handler.mapred;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
import org.wolfengi.handler.common.Constants;
import org.wolfengi.handler.common.ShipCN;
import org.wolfengi.handler.util.StringTools;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Arrays;

/**
 * @Author: wolfengi
 * @Description: ${TODO}
 * @Date: Created in 10:13 2020/5/18
 * @Modified By:
 */
public class FaultMapper {
    private static final Logger log = LoggerFactory.getLogger(FaultMapper.class);

    private static final String DICT = Constants.DICTNAME;
    private static final String DATA = Constants.DATANAME;

    /**
     * 诊断故障船型
     * TODO: 是否清除不在ShipCN内的故障信息；&& null != ShipCN.getCNTrim(fault)
     */
    public static class ShipTypeCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            log.info("[-] executing ShipTypeCountMapper.");
            String data = value.toString();
            if (isNotFileName(context, DATA, data)) {
                return;
            }

            String[] words = data.split("\t");
            String mmsi = words[0].trim();
            String sdate = words[1].trim();
            String year = "";
            String fault = words[words.length - 1].replaceAll("\"", "").trim();
            if (!StringTools.isNull(sdate) && !StringTools.isNull(mmsi) && !StringTools.isNull(fault)) {
                year = sdate.substring(0, sdate.indexOf("/")).trim();
                context.write(new Text(year + "@" + mmsi + "@" + fault), new IntWritable(1));
            }
        }
    }

    /**
     * 诊断故障原因
     */
    public static class ShipFaultCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String data = value.toString();
            if (isNotFileName(context, DATA, data)) {
                return;
            }
            log.info("[-] executing ShipFaultCountMapper.");
            String[] words = data.split("\t");
            String fault = words[words.length - 1].replaceAll("\"", "").trim();
            if (!StringTools.isNull(fault)) {
                context.write(new Text(fault), new IntWritable(1));
            }
        }
    }

    /**
     * 诊断船龄
     */
    public static class ShipAgeCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String data = value.toString();
            if (isNotFileName(context, DATA, data)) {
                return;
            }
            log.info("[-] executing ShipAgeCountMapper.");
            String[] words = data.split("\t");
            String mmsi = words[0].trim();
            if (!StringTools.isNull(mmsi)) {
                context.write(new Text(mmsi), new IntWritable(1));
            }
        }
    }

    /**
     * 诊断故障原因下的不同船型
     */
    public static class ShipFaultTypeCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String data = value.toString();
            if (isNotFileName(context, DATA, data)) {
                return;
            }
            log.info("[-] executing ShipFaultTypeCountMapper.");
            String[] words = data.split("\t");
            String mmsi = words[0].trim();
            String fault = words[words.length - 1].replaceAll("\"", "").trim();
            if (!StringTools.isNull(mmsi) && !StringTools.isNull(fault)) {
                context.write(new Text(mmsi + "@" + fault), new IntWritable(1));
            }
        }
    }

    /**
     * 出行率
     */
    public static class ShipTravelRateCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String data = value.toString();
            if (isNotFileName(context, DATA, data)) {
                return;
            }
            log.info("[-] executing ShipTravelRateCountMapper.");
            String[] words = data.split("\t");
            String mmsi = words[0].trim();
            String date = words[1].trim();
            if (!StringTools.isNull(date) && !StringTools.isNull(mmsi)) {
                DateTimeFormatter dt = DateTimeFormat.forPattern("yyyy/MM/dd");
                String dateTime = DateTime.parse(date, dt).toString("YYYY-MM");
                String state = words[words.length - 6].trim();
                if (ShipCN.STATE_UNDERWAY.getShipEN().equals(state)) {
                    context.write(new Text(mmsi + "@" + dateTime + "@" + state), new IntWritable(1));
                }
            }
        }
    }


    /**
     * 诊断故障与状态
     */
    public static class ShipGroupCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String data = value.toString();
            if (isNotFileName(context, DATA, data)) {
                return;
            }
            log.info("[-] executing ShipGroupCountMapper.");
            String[] words = data.split("\t");
            String state = words[words.length - 6].trim();
            String fault = words[words.length - 1].replaceAll("\"", "").trim();
            if (!StringTools.isNull(state) && !StringTools.isNull(fault)) {
                if (ShipCN.FAULT_NULL.getShipEN().equals(fault)) {
                    context.write(new Text(state + "@" + ShipCN.NORMAL.getShipEN()), new IntWritable(1));
                } else {
                    context.write(new Text(state + "@" + ShipCN.FAULT.getShipEN()), new IntWritable(1));
                }
            }
        }
    }

    /**
     * 概览
     */
    public static class ShipOverviewMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String data = value.toString();
            if (isNotFileName(context, DATA, data)) {
                return;
            }
            log.info("[-] executing ShipOverviewMapper.");
            String[] words = data.split("\t");
            String fault = words[words.length - 1].replaceAll("\"", "").trim();
            if (!StringTools.isNull(fault)) {
                if (!ShipCN.FAULT_NULL.getShipEN().equals(fault)) {
                    context.write(new Text("fault"), new IntWritable(1));
                }
                context.write(new Text("ships"), new IntWritable(1));
            }
        }
    }


    /**
     * 船舶mmsi字典数据清洗入库
     */
    public static class ShipDictMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String data = value.toString();
            if (isNotFileName(context, DICT, data)) {
                return;
            }
            log.info("[-] executing ShipDictMapper.");
            String[] split = data.split("\t");
            if (split.length >= 4) {
                String mmsi = split[0].trim();
                float weight = Float.parseFloat(split[1].trim());
                int age = Integer.parseInt(split[2].trim());
                String type = split[3].trim();
                String k = mmsi + "@" + weight + "@" + age + "@" + type;
                context.write(new Text(k), new IntWritable(1));
            }
        }
    }

    /**
     * 后期实现，船舶年龄清洗；未使用
     */
    public static class ShipAgeBakMapper extends Mapper<LongWritable, Text, Text, Text> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String data = value.toString();
            if (isNotFileName(context, DATA, data)) {
                return;
            }
            FileSplit inputSplit = (FileSplit) context.getInputSplit();
            String filename = inputSplit.getPath().toString();
            String[] split = data.split("\t");
            if (filename.endsWith(DICT)) {
                if (split.length > 2) {
                    context.write(new Text(split[0]), new Text(String.valueOf(split[2])));
                }
            }
            if (filename.endsWith(DATA)) {
                if (split.length > 1) {
                    context.write(new Text(split[0]), new Text(split[0]));
                }
            }
        }
    }

    /**
     * 判断是否是要处理的文件，包含脏数据处理
     *
     * @param context
     * @param fileName
     * @return
     */
    private static boolean isNotFileName(Mapper.Context context, String fileName, String columns) {
        FileSplit inputSplit = (FileSplit) context.getInputSplit();
        String filename = inputSplit.getPath().toString();
        if (!filename.endsWith(fileName)) {
            return true;
        }
        if (isNONE(filename, columns)) {
            return true;
        }
        return false;
    }


    /**
     * TODO：脏数据处理
     * 文件名不是 dict.txt & data.txt
     * 文件内容包含空格，空字符串
     * 状态 和 故障类型 在该对应的类型范围之外
     * 则 返回 true
     *
     * @param filename
     * @param columns
     * @return
     */
    private static boolean isNONE(String filename, String columns) {
        columns = columns.replaceAll("\"", "");
        String[] split = columns.split("\t");
        if (filename.endsWith(DICT)) {
            /**
             * TODO: 字典表：当 column 的数量为4时，说明数据正常，进行处理
             */
            if (split.length == 4) {
                for (String s : Arrays.asList(split)) {
                    if (StringTools.isNull(s)) {
                        /**
                         * 为空时，返回true，说明数据不对，会进入上层if条件，不通过后面清洗步骤
                         */
                        return true;
                    }
                }
                if (null == ShipCN.getCNTrim(split[split.length - 1].trim())) {
                    return true;
                }
                return false;
            }
        }
        if (filename.endsWith(DATA)) {
            /**
             * TODO: 数据表：当 column 的数量为4时，说明数据正常，进行处理
             */
            if (split.length == 18) {
                for (String s : Arrays.asList(split)) {
                    if (StringTools.isNull(s)) {
                        /**
                         * 为空时，返回true，说明数据不对，会进入上层if条件，不通过后面清洗步骤
                         */
                        return true;
                    }
                }
                if (null == ShipCN.getCNTrim(split[split.length - 1].trim()) || null == ShipCN.getCNTrim(split[split.length - 6].trim())) {
                    return true;
                }
                return false;
            }
        }

        return true;
    }
}
